package com.gitee.xhs.savePointReader.keyedState.reducingStateReader;

import com.gitee.xhs.keyedState.reducingState.ReducingStateConstant;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.state.api.functions.KeyedStateReaderFunction;
import org.apache.flink.util.Collector;

/**
 * @author BigHead
 * @date 2022/6/30 14:09
 * @description ReducingStateReaderFunction
 */
public class ReducingStateReaderFunction extends KeyedStateReaderFunction<Long, ReducingStateExhibitor> {

    private final String jobId;
    private final String operatorUid;
    private transient ReducingState<Double> reducingState;

    public ReducingStateReaderFunction(String jobId, String operatorUid) {
        this.jobId = jobId;
        this.operatorUid = operatorUid;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        reducingState = getRuntimeContext().getReducingState(ReducingStateConstant.REDUCING_STATE_DESCRIPTOR);
    }

    @Override
    public void readKey(Long key, Context ctx, Collector<ReducingStateExhibitor> out) throws Exception {
        ReducingStateExhibitor reducingStateExhibitor = new ReducingStateExhibitor();
        reducingStateExhibitor.setJobId(jobId);
        reducingStateExhibitor.setOperatorUid(operatorUid);
        reducingStateExhibitor.setKey(key);
        reducingStateExhibitor.setReducingStateValue(reducingState.get());

        out.collect(reducingStateExhibitor);
    }
}
